package com.lagou.zk;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;

import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.*;

/**
 * ZK工具类
 */
@Slf4j
public class ZkHelper {
    public static final String SVR_REG_NODE_PATH = "/server";
    public static final String SVR_SPEED_PATH = "/server-speed";
    public static final String SVR_NODE_PREFIX = "svr-";
    private static final int SVR_PORT_START = 9000;
    private static final List<PathChildrenCacheEvent.Type> LISTEN_TYPES = Arrays.asList(CHILD_ADDED, CHILD_UPDATED, CHILD_REMOVED);
    private static CuratorFramework client = null;
    private static int SVR_CAPACITY = 1000;

    /**
     * 从ZK状态通知中获取服务端信息
     *
     * @param childData
     * @return
     */
    private static ServerInfoVO getServerInfo(ChildData childData) {
        String data = new String(childData.getData(), StandardCharsets.UTF_8);
        log.info(data);
        ServerInfoVO serverInfo = JSON.parseObject(data, ServerInfoVO.class);
        return serverInfo;
    }

    /**
     * 获取ZK客户端
     *
     * @return
     * @throws Exception
     */
    public static CuratorFramework getClient() throws Exception {
        //客户端不存在时，初始化连接
        if (client == null) {
            initClient();
        }
        return client;
    }

    /**
     * 注册服务信息，由ZK分配端口，但服务处于未启动状态，客户端还不可使用
     *
     * @param ip
     * @throws Exception
     */
    public static ServerInfoVO regServer(String ip) throws Exception {
        // 构造服务信息
        ServerInfoVO infoVO = new ServerInfoVO();
        infoVO.setIp(ip);
        infoVO.setIsUp(false);
        byte[] infoBytes = JSON.toJSONBytes(infoVO);
        // 注册顺序临时节点，监控服务状态同时，分配唯一服务名
        // 该操作应该用其它节点进行，模拟项目简化，直接使用监听路径，用isUp来标识状态
        String nodeName = getClient().create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(String.format("%s/%s", SVR_REG_NODE_PATH, SVR_NODE_PREFIX), infoBytes);
        // 取服务名序列，与服务端容量取模后，获取服务端口
        String nodeSeq = nodeName.substring(nodeName.indexOf("-") + 1);
        int nodeIdx = Integer.parseInt(nodeSeq);
        int portHash = nodeIdx % SVR_CAPACITY;
        // 此处需考虑端口占用，以及循环后端口重叠，模拟项目忽略
        int port = SVR_PORT_START + portHash;
        // 更新服务器名、端口
        infoVO.setPort(port);
        infoVO.setServerName(nodeName);
        return infoVO;
    }

    /**
     * 更新最后响应时间
     *
     * @param serverName
     * @param speed
     * @throws Exception
     */
    public static void updateServerSpeed(String serverName, long speed) throws Exception {
        // 根据服务节点名，获得速度节点路径
        String nodeName = serverName.substring(serverName.lastIndexOf("/") + 1);
        String speedNode = String.format("%s/%s", SVR_SPEED_PATH, nodeName);
        // 创建数据记录节点
        // 原计划通过TTL实现5秒归0，但实践后ZK的TTL实现不如预期且有版本和配置要求，因此放弃并采用客户端去判断并清0
        getClient().create().orSetData()
                // 设置5秒超时
                // .withTtl(1)
                .withMode(CreateMode.EPHEMERAL)
                .forPath(speedNode, (speed + "").getBytes());
        log.info("更新节点[{}]速度[{}]", speedNode, speed);
    }

    /**
     * 服务监听后，更新服务状态为可用
     *
     * @param serverInfo
     * @throws Exception
     */
    public static void serverUp(ServerInfoVO serverInfo) throws Exception {
        serverInfo.setIsUp(true);
        byte[] infoBytes = JSON.toJSONBytes(serverInfo);
        getClient().setData().forPath(serverInfo.getServerName(), infoBytes);
    }

    /**
     * 主动清除服务信息
     *
     * @param serverInfo
     * @throws Exception
     */
    public static void serverDown(ServerInfoVO serverInfo) throws Exception {
        getClient().delete().forPath(String.format("%s/%s", SVR_REG_NODE_PATH, serverInfo.getServerName()));
    }

    /**
     * 检测服务注册节点，不存在则创建
     *
     * @throws Exception
     */
    public static void createRegNodeIfNotExists(String path) throws Exception {
        var existsNode = getClient().checkExists().forPath(path);
        if (existsNode == null) {
            getClient().create().forPath(path);
            log.info("创建节点 [{}]", path);
        }
    }

    /**
     * 初始化ZK客户端，通过虚拟机同步方法防止重复初始化
     *
     * @throws Exception
     */
    private synchronized static void initClient() throws Exception {
        ExponentialBackoffRetry retry = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory
                .builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(1000)
                .connectionTimeoutMs(10000)
                .retryPolicy(retry)
                .namespace("task2")
                .build();
        client.start();
        // 客户端单例
        ZkHelper.client = client;
        // 检测服务注册节点，不存在则创建
        createRegNodeIfNotExists(SVR_REG_NODE_PATH);
        // 检测服务速度节点，不存在则创建
        createRegNodeIfNotExists(SVR_SPEED_PATH);
    }

    /**
     * 监听服务器变化，该方法供客户端使用
     *
     * @param handler
     * @throws Exception
     */
    public static void watchServerNode(ServerInfoHandler handler) throws Exception {
        // 设置监听路径
        PathChildrenCache pathChildrenCache = new PathChildrenCache(ZkHelper.getClient(), SVR_REG_NODE_PATH, true);
        // 设置监听方法
        pathChildrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {
            log.info("SERVER >> {}", pathChildrenCacheEvent.getType().toString());
            // 只监听 ADD REMOVE UPDATE 方法
            if (LISTEN_TYPES.contains(pathChildrenCacheEvent.getType())) {
                ServerInfoVO serverInfo = getServerInfo(pathChildrenCacheEvent.getData());
                switch (pathChildrenCacheEvent.getType()) {
                    case CHILD_ADDED:
                        if (serverInfo.getIsUp()) {
                            // 通知客户端有新服务器上线
                            handler.onServerAdd(serverInfo);
                        }
                        break;
                    case CHILD_UPDATED:
                        if (serverInfo.getPort() != null) {
                            // 通知客户端有服务器状态变更
                            handler.onServerUpdate(serverInfo);
                        }
                        break;
                    case CHILD_REMOVED:
                        // 通知客户端有服务器下线
                        handler.onServerRemove(serverInfo);
                        break;
                }

            }
        });
        pathChildrenCache.start();
    }

    /**
     * 监听服务器变化，该方法供客户端使用
     *
     * @param handler
     * @throws Exception
     */
    public static void watchServerSpeed(ServerSpeedHandler handler) throws Exception {
        // 设置监听路径
        PathChildrenCache pathChildrenCache = new PathChildrenCache(ZkHelper.getClient(), SVR_SPEED_PATH, true);
        // 设置监听方法
        pathChildrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {
            String path = pathChildrenCacheEvent.getData().getPath();
            // 只监听 ADD REMOVE UPDATE 方法
            if (LISTEN_TYPES.contains(pathChildrenCacheEvent.getType())) {
                String serverName = String.format("%s/%s", SVR_REG_NODE_PATH, path.substring(path.lastIndexOf("/") + 1));
//                log.info("SERVER SPEED {} >> {}", pathChildrenCacheEvent.getData().getPath(), pathChildrenCacheEvent.getType().toString());
                String speedInfo = new String(pathChildrenCacheEvent.getData().getData());
                long speed = Long.parseLong(speedInfo);
                switch (pathChildrenCacheEvent.getType()) {
                    case CHILD_ADDED:
                        handler.onServerSpeedAdd(serverName, speed);
                        break;
                    case CHILD_REMOVED:
                        handler.onServerSpeedRemove(serverName);
                        break;
                    case CHILD_UPDATED:
                        handler.onServerSpeedUpdate(serverName, speed, System.currentTimeMillis());
                        break;
                }

            }
        });
        pathChildrenCache.start();
    }
}
